package cn.antcore.kafka.builder;

import cn.antcore.kafka.annotation.KafkaClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * Created By Hong on 2018/8/21
 **/
public class KafkaBuilder {

    private static final Logger LOG = LoggerFactory.getLogger(KafkaBuilder.class);

    /**
     * Bean工厂
     */
    private static ConfigurableListableBeanFactory BEAN_FACTORY;

    /**
     * 缓存的KafkaClient
     */
    private static Map<String, Object> CLIENTS = new HashMap<>();

    /**
     * 创建Client
     *
     * @param beanFactory
     * @param names       KafkaClient 名字
     */
    public static void createClient(ConfigurableListableBeanFactory beanFactory, String[] names) {
        KafkaBuilder.BEAN_FACTORY = beanFactory;
        new KafkaBuilder(names);
    }

    /**
     * 构造函数
     *
     * @param names KafkaClient 名字
     */
    private KafkaBuilder(String[] names) {
        if (names == null) {
            if (LOG.isInfoEnabled()) {
                LOG.info("KafkaClient is empty.");
            }
            return;
        }
        createClient(names);
    }

    /**
     * 创建Client
     *
     * @param names KafkaClient 名字
     */
    private void createClient(String[] names) {
        for (String name : names) {
            if (!CLIENTS.containsKey(name)) {
                Object target = BEAN_FACTORY.getBean(name);
                if (AopUtils.isAopProxy(target)) {
                    target = getTargetObject(target);
                }
                //获取注解
                KafkaClient client = AnnotationUtils.findAnnotation(target.getClass(), KafkaClient.class);
                ConsumerBuilder consumerBuilder = ConsumerBuilder.create(target, client.topic(), client.groupId(), client.bootstrapService(), client.config());
                CLIENTS.put(name, consumerBuilder.build());
            }
        }
    }


    /**
     * 获取aop代理的目标对象
     *
     * @param candidate 源对象
     * @param <T>       目标对象
     * @return 目标对象
     */
    private static <T> T getTargetObject(Object candidate) {
        try {
            if (AopUtils.isAopProxy(candidate) && (candidate instanceof Advised)) {
                return (T) ((Advised) candidate).getTargetSource().getTarget();
            }
        } catch (Exception ex) {
            throw new IllegalStateException("Failed to unwrap proxied object", ex);
        }
        return (T) candidate;
    }
}
